package com.study.kafka;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.KafkaFuture;

import java.util.ArrayList;
import java.util.Properties;
import java.util.Set;

/**
 * @author liuguibin
 * @date 2021-01-15 15:31
 * 验证是否连接成功
 */
public class kafkaConnTest {
    public static String get() {
        AdminClient kafkaConn = null;
        try {
            Properties prop = new Properties();
            prop.put("bootstrap.servers", "188.188.23.161:9092");
            prop.put("retries", 0);
            prop.put("request.timeout.ms", 1000);
            kafkaConn = AdminClient.create(prop);
            //创建新的topic
            ArrayList<NewTopic> newTopics = new ArrayList<>();
            NewTopic newTopic = new NewTopic("topic-test2", 1, (short) 1);
            newTopics.add(newTopic);

            //验证topic 是否存在
            ListTopicsResult listTopicsResult = kafkaConn.listTopics();
            KafkaFuture<Set<String>> topics = listTopicsResult.names();
            Set<String> strings = topics.get();
            if (strings.size() > 0) {
                return "成功";
            } else {
                //不存在topic,创建topic
                try {
                    CreateTopicsResult result = kafkaConn.createTopics(newTopics);
                    return "成功";
                } catch (Exception e) {
                    e.printStackTrace();
                    return "失败";
                }
            }

        } /*catch (TimeoutException e) {
            e.printStackTrace();
            return "失败";
        } catch (InterruptedException e) {
            e.printStackTrace();
            return "失败";
        } catch (ExecutionException e) {
            e.printStackTrace();
            return "失败";
        }*/catch (Exception e){
            e.printStackTrace();
            return "失败";
        }finally {
            kafkaConn.close();
        }
    }

    public static void main(String[] args) {
        String s = kafkaConnTest.get();
        System.out.println(s);

    }
}
